package com.pw.study.flink.source


import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet, ResultSetMetaData}
import java.util.Random

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.source.{RichSourceFunction, SourceFunction}

class CustomSourceByRich extends RichSourceFunction[UserPO] {
  var isSwitch = true
  var connection: Connection = null
  var statement: PreparedStatement = null

  override def open(parameters: Configuration): Unit = {
    print("open...............")
    Class.forName("com.mysql.jdbc.Driver")
    connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/study", "root", "123456")
    val sql: String = "select * from user"
    statement = connection.prepareStatement(sql)


  }

  override def run(ctx: SourceFunction.SourceContext[UserPO]): Unit = {
    while (isSwitch) {

      val resultSet = statement.executeQuery()
      val metaData: ResultSetMetaData = resultSet.getMetaData
      val count: Int = metaData.getColumnCount
      while (resultSet.next()) {
        val id: String = resultSet.getString(1)
        val name: String = resultSet.getString(2)
        val age: Int = resultSet.getInt(3)
        val email: String = resultSet.getString(4)
        println("id:" + id + " name: " + name+" age: "+age+" email: "+email)
        val userPO = UserPO(id, name, age, email)
        ctx.collect(userPO)
        println("生成一条数据：" + userPO)
      }
      Thread.sleep(1000*60*60)
    }
  }

  override def cancel(): Unit = {
    isSwitch = false
  }
}

case class UserPO(id: String, name: String, age: Int, email: String)